import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.security.KeyFactory;
import java.security.NoSuchAlgorithmException;
import java.security.NoSuchProviderException;
import java.security.PublicKey;
import java.security.spec.InvalidKeySpecException;
import java.security.spec.X509EncodedKeySpec;
import java.util.*;
import java.net.*;

public class Peer 
{

    /**
     * @param args
     */
    PeerKeyPair myKeyPair;
    ArrayList<PublicKey> otherKeyPair;
    ArrayList<PublicKey> clientPublicKey;
    String logFile;
    String pubKeyFile;
    DatagramSocket pSocket;
    int pid;
    int port;
    int view;
    int totalPeers;
    String role;
    int primaryPid;
    ArrayList<Integer> replicaUpSet;
    ArrayList<Integer> clientList;
    ArrayList<String> lastClientResponse;
    String peerType;
    int currSequenceNum;
    ArrayList<String> listPrePrepareAccepted;
    ArrayList<String> listPrepareAccepted;
    ArrayList<String> listViewChangeAccepted;
    ArrayList<String> listReplyAccepted;
    ArrayList<String> listLog;
    ArrayList<String> listCheckpointAccepted;
    ArrayList<Payload> listClientPayload;
    String logItem;
    Payload viewChangePayload;
    int highWaterMark;
    int lowWaterMark;
	private ArrayList<String> listCommitAccepted;
	int lastStableCheckpoint;
    int peerTimeout;
    int clientAllReqIndx;
    boolean waitingInViewChange;
    boolean isByzantine;
    
    public Peer(int pid, int port,String peerType, boolean isByzantine,int peerTimeout,ArrayList<Integer> upSet) 
    {
    	this.isByzantine = isByzantine;
        this.pid = pid;
        this.port = port;
        this.view = 0;
        this.currSequenceNum=0;
        this.peerType = peerType;
        this.logFile = this.peerType+this.pid+".log"; 
        this.pubKeyFile = this.peerType+this.pid+".pubKey";
        this.replicaUpSet = upSet;
        this.primaryPid = Constants.MINUS_ONE;
        this.peerTimeout = peerTimeout;
        this.clientAllReqIndx = 0;
        this.waitingInViewChange = false;
        System.out.println("");
        myKeyPair = Utilities.generateMyKeys();
        totalPeers = upSet.size();
        if (peerType.equals(Constants.CLIENT))
        {
                this.role = Constants.CLIENT;
        }
        else if(peerType.equals(Constants.PEER))
        {
                this.role = Constants.BACKUP;
        }
        clientPublicKey = new ArrayList<PublicKey>();
        otherKeyPair = new ArrayList<PublicKey>();
        listPrePrepareAccepted = new ArrayList<String>();
        listPrepareAccepted = new ArrayList<String>();
        listCommitAccepted = new ArrayList<String>();
        clientList = new ArrayList<Integer>();
        listReplyAccepted = new ArrayList<String>();
        listLog = new ArrayList<String>();
        lastClientResponse = new ArrayList<String>();
        listViewChangeAccepted = new ArrayList<String>();
        listCheckpointAccepted = new ArrayList<String>();
        listClientPayload = new ArrayList<Payload>();
        viewChangePayload = new Payload();
        try 
        {
                pSocket = new DatagramSocket(port);
              
        } 
        catch (SocketException se) 
        {
                // TODO: handle exception
                se.printStackTrace();
        }
        try 
        {
                new FileOutputStream(this.pubKeyFile).write(this.myKeyPair.pubKey.getEncoded());
        } 
        catch (FileNotFoundException e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
        } 
        catch (IOException e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
        }

               
        this.highWaterMark = Constants.INIT_WATERMARK;
        this.lowWaterMark = 0;
        this.lastStableCheckpoint = 0;
    }
    
    public PublicKey addPublicKey(String peerType, int peerId)
    {
        PublicKey peerPublicKey = null;
/*        if (peerType.equals(Constants.CLIENT))
        {
            try 
            {
                peerPublicKey = clientPublicKey.get(peerId);
                return peerPublicKey;
            } catch (IndexOutOfBoundsException ioobe) 
            {
                //ioobe.printStackTrace();
            }
        }
        else if(peerType.equals(Constants.PEER))
        {
            try 
            {
                peerPublicKey = otherKeyPair.get(peerId);
                return peerPublicKey;
            } 
            catch (IndexOutOfBoundsException ioobe) 
            {
                //ioobe.printStackTrace();
            }

        }*/
        String peerPubKeyFile = peerType+peerId+".pubKey";
        byte[] encodedKey = new byte[(int)new File(peerPubKeyFile).length()];
        try 
        {
            new FileInputStream(peerPubKeyFile).read(encodedKey);
        } 
        catch (FileNotFoundException e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
        }
        catch (IOException io) 
        {
                // TODO Auto-generated catch block
                io.printStackTrace();
        }
        X509EncodedKeySpec pubKeySpec = new X509EncodedKeySpec(encodedKey);
        KeyFactory keyFactory = null;
        try 
        {
            keyFactory = KeyFactory.getInstance(Constants.keyAlgo, Constants.provider);
        }
        catch (NoSuchAlgorithmException nsae) 
        {
            // TODO Auto-generated catch block
            nsae.printStackTrace();
        }
        catch (NoSuchProviderException nspe) 
        {
            // TODO Auto-generated catch block
            nspe.printStackTrace();
        }
        try 
        {
            peerPublicKey = keyFactory.generatePublic(pubKeySpec);
        } 
        catch (InvalidKeySpecException ikse) 
        {
            // TODO Auto-generated catch block
            ikse.printStackTrace();
        }
/*        if(peerType.equals(Constants.CLIENT))
        {
            clientPublicKey.add(peerPublicKey);
        }
        else if(peerType.equals(Constants.PEER))
        {
            otherKeyPair.add(peerPublicKey);
        }
                */
        return peerPublicKey;
    
    }
    
	public boolean validatePrePrepare(Payload incomingPayload)
	{
		//verify primary's signature
		
		boolean primaryVerified,clientVerified;
		PublicKey primaryPublicKey;
		primaryPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
        primaryVerified = Utilities.verifySignedData(incomingPayload, primaryPublicKey);
	
		if(!primaryVerified)
		{
			System.out.println("PID: "+pid+ " - Primary signature is NOT VALID!!!!!!");
			return false;
		}
		//verify sign in client req
		System.out.println("PID: "+pid+ " - Primary signature is valid");
		Payload clientPayload = incomingPayload.clientPayload;
		PublicKey clientPublicKey = addPublicKey(Constants.CLIENT,clientPayload.senderPid);
		clientVerified = Utilities.verifySignedData(clientPayload, clientPublicKey);
		// TODO: digest comparison
		if(!clientVerified)
		{
			System.out.println("PID: "+pid+ " - Client signature is NOT VALID!!!!!!");
			return false;
		}
		
		System.out.println("PID: "+pid+ " - Client signature is valid");
		if (!Utilities.verifyHash(clientPayload.payloadByteArray(), incomingPayload.digest))
		{
			System.out.println("PID: "+pid+ " - Message hash is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Message hash is valid");
		if(!(view == incomingPayload.view))
		{
			
			System.out.println("PID: "+pid+ " - Incoming view is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming view is valid");
		if(!(incomingPayload.sequenceNum <= highWaterMark && incomingPayload.sequenceNum >= lowWaterMark))
		{
			System.out.println("PID: "+pid+ " - Incoming sequnce number is NOT VALID!!!!!!");
			return false;
		}
		
		System.out.println("PID: "+pid+ " - Incoming sequence number is valid");
		String prePrepareCheck = view+""+incomingPayload.sequenceNum+incomingPayload.digest;
		if(listPrePrepareAccepted.indexOf(prePrepareCheck) != -1)
		{
			return false;
		}

		for(int i=0;i < listPrePrepareAccepted.size();i++)
		{
			if(listPrePrepareAccepted.get(i).substring(0, 2).equals(view+""+incomingPayload.sequenceNum))
			{
				System.out.println("PID: "+pid+ " - Got PRE-PREPARE with same sequence number and different digest in current view. Rejecting!");
				return false;
			}
		}
				
		listPrePrepareAccepted.add(prePrepareCheck);
		
		System.out.println("PID: "+pid+ " - All criteria for incoming PRE-PREPARE satisfied. Accepting and logging PRE-PREPARE");
		logItem = (incomingPayload.sequenceNum)+","+Constants.PREPREPARE+","+view+","+Utilities.hashMyData(incomingPayload.payloadByteArray());
		listLog.add(logItem);
		writeLog(listLog);
		return true;
	}	


        
	public boolean validatePrepare(Payload incomingPayload)
	{
		//verify replica's signature
		//	//PREPARE log format: "PREPARE, view, seq, digest,myPid"
		boolean replicaVerified;
		PublicKey replicaPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
		replicaVerified = Utilities.verifySignedData(incomingPayload, replicaPublicKey);
		if(!replicaVerified)
		{
			System.out.println("PID: "+pid+ " - Replica signature is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Replica signature is valid");
		if(!(view == incomingPayload.view))
		{
			System.out.println("PID: "+pid+ " - Incoming view is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming view is valid");
		if(!(incomingPayload.sequenceNum <= highWaterMark && incomingPayload.sequenceNum >= lowWaterMark))
		{
			System.out.println("PID: "+pid+ " - Incoming sequence number is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming sequence number is valid");
		
		System.out.println("PID: "+pid+ " - All criteria for incoming PREPARE satisfied. Accepting and logging PREPARE message");
		String newPrepareAccepted = incomingPayload.senderPid+""+view+incomingPayload.sequenceNum+incomingPayload.digest;
		if(listPrepareAccepted.indexOf(newPrepareAccepted) != -1)
		{
			return false;
		}
		listPrepareAccepted.add(newPrepareAccepted);
		
		listLog.add(incomingPayload.sequenceNum+","+Constants.PREPARE+","+incomingPayload.view+","+incomingPayload.digest+","+pid);
		writeLog(listLog);
		return true;
	}
	
	public boolean validateCommit(Payload incomingPayload) 
	{
		//verify replica's signature
		boolean replicaVerified;
		PublicKey replicaPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
		replicaVerified = Utilities.verifySignedData(incomingPayload, replicaPublicKey);
		if(!replicaVerified)
		{
			System.out.println("PID: "+pid+ " - Replica signature is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Replica signature is valid");
		if(!(view == incomingPayload.view))
		{
			System.out.println("PID: "+pid+ " - Incoming view is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming view is valid");
		if(!(incomingPayload.sequenceNum <= highWaterMark && incomingPayload.sequenceNum >= lowWaterMark))
		{
			System.out.println("PID: "+pid+ " - Incoming sequence number is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming sequence number is valid");
		
		System.out.println("PID: "+pid+ " - All criteria for incoming COMMIT are satisfied. Accepting COMMIT and logging.");
		String newCommitAccepted = incomingPayload.senderPid+""+view+incomingPayload.sequenceNum+incomingPayload.digest; 
		if(listCommitAccepted.indexOf(newCommitAccepted) != -1)
		{
			return false;
		}
		listCommitAccepted.add(newCommitAccepted);
		listLog.add(incomingPayload.sequenceNum+","+Constants.COMMIT+","+incomingPayload.view+","+incomingPayload.digest+","+pid);
		writeLog(listLog);
		return true;
	}
	
	public boolean validateViewChange(Payload incomingPayload) 
	{
		//verify replica's signature
		
		boolean replicaVerified;
		PublicKey replicaPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
		replicaVerified = Utilities.verifySignedData(incomingPayload, replicaPublicKey);
		if(!replicaVerified)
		{
			System.out.println("PID: "+pid+ " - Replica signature is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Replica signature is valid");
		if(!(view >= incomingPayload.view))
		{
			System.out.println("PID: "+pid+ " - Incoming view is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming view is valid");
		if(incomingPayload.listPrepareProof == null)
		{
			System.out.println("PID: "+pid+ " - Proof for PREPAREs is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - All criteria for incoming VIEWCHANGE satisfied. Accepting VIEWCHANGE and logging it");
		String newViewChangeAccepted = incomingPayload.senderPid+""+view+""+incomingPayload.lastStableCheckpoint; 
		if(listViewChangeAccepted.indexOf(newViewChangeAccepted) != -1)
		{
			return false;
		}
		listViewChangeAccepted.add(newViewChangeAccepted);
		listLog.add(incomingPayload.sequenceNum+","+Constants.VIEWCHANGE+","+incomingPayload.view+","+pid);
		writeLog(listLog);
		return true;
	}

	public boolean validateReply(Payload incomingPayload,String requestSent,long reqTimestamp) 
	{
		//verify replica's signature
		boolean replicaVerified;
		
		PublicKey replicaPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
		replicaVerified = Utilities.verifySignedData(incomingPayload, replicaPublicKey);
		if(!replicaVerified)
		{
			System.out.println("PID: "+pid+ " - Replica signature is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Replica signature is valid");
		if(incomingPayload.clientPayload.timestamp != reqTimestamp)
		{
			System.out.println("PID: "+pid+ " - Incoming timestamp is NOT VALID!!!!!!");
			return false;
		}
		System.out.println("PID: "+pid+ " - Incoming timestamp is valid");
		System.out.println("PID: "+pid+ " - Incoming REPLY message is validated");		

		String newReplyAccepted = incomingPayload.senderPid+""+reqTimestamp+""+requestSent+incomingPayload.reply;
		if (listReplyAccepted.indexOf(newReplyAccepted) != -1)
		{
			return false;
		}
		listReplyAccepted.add(newReplyAccepted);
		listLog.add(Constants.REPLY+","+incomingPayload.view+","+incomingPayload.senderPid+","+incomingPayload.reply);
		writeLog(listLog);
		return true;
	}

	public boolean validateCheckpoint(Payload incomingPayload)
       {
    	   boolean replicaVerified;
           PublicKey replicaPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
           replicaVerified = Utilities.verifySignedData(incomingPayload, replicaPublicKey);
           if(!replicaVerified)
           {	
        	   System.out.println("PID: "+pid+ " - Replica signature is NOT VALID!!!!!!");
                   return false;
           }
           System.out.println("PID: "+pid+ " - Replica signature is valid");
           
           System.out.println("PID: "+pid+ " - All criteria for CHECKPOINT message satisfied. Accepting it.");
           String newCheckpointAccepted = incomingPayload.senderPid+""+incomingPayload.sequenceNum+""+incomingPayload.digest;
           if(listCheckpointAccepted.indexOf(newCheckpointAccepted) != -1)
           {
                   return false;
           }
           listCheckpointAccepted.add(newCheckpointAccepted);
           listLog.add(incomingPayload.sequenceNum+","+Constants.CHECKPOINT+","+incomingPayload.view+","+incomingPayload.digest+","+pid);
   		   writeLog(listLog);
           return true;
       }

	public boolean validateNewView(Payload incomingPayload)
	{
		 boolean replicaVerified;
		 
	     PublicKey replicaPublicKey=addPublicKey(Constants.PEER, incomingPayload.senderPid);
	     replicaVerified = Utilities.verifySignedData(incomingPayload, replicaPublicKey);
	     if(!replicaVerified)
	     {
	    	 System.out.println("PID: "+pid+ " - Replica signature is NOT VALID!!!!!!");
	            return false;
	     }
	     System.out.println("PID: "+pid+ " - Replica signature is valid");
	     if(incomingPayload.listViewChangeAccepted == null)
	     {
	    	 System.out.println("PID: "+pid+ " - Set of VIEWCHANGE messages is NOT VALID!!!!!!");
	    	 return false;
	     }
	    	 
	     
	     ArrayList<String> listIncomingViewChanges = new ArrayList<String>(incomingPayload.listViewChangeAccepted);
	     
	     int countOfView = 0;
	     
	     for(int i=0;i<listIncomingViewChanges.size();i++)
	     {
	    	 if(Integer.parseInt(listIncomingViewChanges.get(i).substring(1, 2)) == incomingPayload.view)
	    	 {
	    		 countOfView++;
	    	 }
	     }
	     
	     if(countOfView < 2*Constants.FAULT_TOL)
	     {
	    	 return false;
	     }
	     System.out.println("PID: "+pid+ " - Incoming NEWVIEW message has 2f valid VIEWCHANGE messages");
	     return true;
	}
	

	public boolean isPrepared(Payload replicaPayload)
	{
		//this is prepare packet containing client payload as well.
		//check in log first if pre-prepare exists for message m, seq and view
		
		
		String prePrepareCheck = view+""+replicaPayload.sequenceNum+replicaPayload.digest;
		
		if (listPrePrepareAccepted.indexOf(prePrepareCheck) == -1)
		{
			System.out.println("PID: "+pid+ " - DID NOT find matching PRE-PREPARE for this PREPARE");
			return false;
		}
		String prepareCheck = prePrepareCheck;
		
		int prepareFound=0;
		
		for (int i=0;i<listPrepareAccepted.size();i++)
		{
			if(listPrepareAccepted.get(i).substring(1).equals(prepareCheck))
			{
				prepareFound++;
			}
		}
		
		if (prepareFound == 2*Constants.FAULT_TOL)
		{
			System.out.println("PID: "+pid+ " - Found 2f matching PREPAREs with same view, sequence number and digest. Prepared predicate is TRUE");
			return true;
		}
		return false;
	}

	public boolean isCommittedLocal(Payload incomingPayload) 
	{
		// TODO Auto-generated method stub
		String prePrepareCheck = view+""+incomingPayload.sequenceNum+incomingPayload.digest;
		String prepareCheck = prePrepareCheck;
		/*if(!isPrepared(incomingPayload))
		{
			return false;
		}*/
		int commitFound=0;
		for (int i=0;i<listCommitAccepted.size();i++)
		{
			if(listCommitAccepted.get(i).substring(1).equals(prepareCheck))
			{
				commitFound++;
			}
		}
		if (commitFound == 2*Constants.FAULT_TOL+1)
		{
			System.out.println("PID: "+pid+ " - Found 2f+1 matching COMMITs that match the PRE-PREPARE. Commited-local predicate is TRUE");
			return true;
		}
		return false;
	}
	
	public boolean isViewChanged(Payload incomingPayload)
	{
		String viewChangeCheck = view+""+incomingPayload.lastStableCheckpoint;
		int viewChangeFound=0;
		for (int i=0;i<listViewChangeAccepted.size();i++)
		{
			if(listViewChangeAccepted.get(i).substring(1).equals(viewChangeCheck))
			{
				viewChangeFound++;
			}
		}
		if (viewChangeFound <= 2*Constants.FAULT_TOL)
		{
			return false;
		}
		
		return true;
	}
	
    public boolean isStable(Payload incomingPayload)
    {
    	String checkpointCheck = incomingPayload.sequenceNum+""+incomingPayload.digest;
        
        int checkpointFound=0;
        for (int i=0;i<listCheckpointAccepted.size();i++)
        {
            if(listCheckpointAccepted.get(i).substring(1).equals(checkpointCheck))
            {
            	checkpointFound++;
            }
        }
       
        if (checkpointFound == 2*Constants.FAULT_TOL + 1)
        {
        	if(incomingPayload.sequenceNum != 0)
        	{
        		System.out.println("PID: "+pid+ " - Got 2f+1 valid CHECKPOINT messages. Checkpoint is stable. Discarding earlier checkpoints and messages");
                return true;
        	}
        }
        return false;
    	
    }
	
	public void processCommit(Payload incomingPayload)
	{
		
		boolean verified = validateCommit(incomingPayload);
		if(!verified)
		{
			return;
		}

		if(!isCommittedLocal(incomingPayload))
		{
			return;
		}
		//now send response to client
		sendResponseToClient(incomingPayload);
		//checkpoint
        if((incomingPayload.sequenceNum % Constants.CHECKPOINT_INT)==0)
        {
        	if(incomingPayload.sequenceNum != 0)
        	{
        		generateCheckpoint(incomingPayload.sequenceNum, incomingPayload.digest);
        	}
        }

	}
	
	public void sendResponseToClient(Payload incomingPayload)
	{
    	String respToClient = incomingPayload.clientPayload.request;
    	Scanner scanInp = new Scanner(System.in);
    	if(isByzantine)
    	{
    		System.out.print("PID: "+pid+ " - Behave Byzantine in client response?: ");
    		String byzBehave = scanInp.next();
    		System.out.println("");
    		if(byzBehave.equals(Constants.YES))
    		{
    			System.out.print("PID: "+pid+ " - Enter response to send to client: ");
    			respToClient = scanInp.next();
    		}
    	}
    	else
    	{
    		System.out.println("PID: "+pid+ " - Sending response "+respToClient+" to client ID " +incomingPayload.clientPayload.senderPid);
    		scanInp.nextLine();
    		
    	}
    	
		byte[] responseData = (new PayloadBuilder()).buildReply(this.view,incomingPayload.sequenceNum,
		this.pid,this.myKeyPair,incomingPayload.digest,incomingPayload.clientPayload,respToClient);

		try
		{
			
			
		    pSocket.send(new DatagramPacket(responseData,responseData.length,
		    		InetAddress.getByName("localhost"),clientList.get(incomingPayload.clientPayload.senderPid)));
		    listLog.add(incomingPayload.sequenceNum+","+Constants.REPLY+","+incomingPayload.view+","+respToClient+","+pid);
	   		writeLog(listLog);
			lastClientResponse.set(incomingPayload.clientPayload.senderPid, respToClient+";"+incomingPayload.clientPayload.request);
		}
		catch (UnknownHostException e) 
		{
			e.printStackTrace();
		}
		catch (IOException e) 
		{
			e.printStackTrace();
		}
		
		
		
	}
	
	 public void generateCheckpoint(int sequenceNum, String digest)
        {
        	byte[] checkpointData = (new PayloadBuilder()).buildCheckpoint(this.view,sequenceNum,
                    this.pid,this.myKeyPair,digest);

            try
            {
                    System.out.println("PID: "+pid+" - Sending checkpoint messages to all");
                    for( int i=0; i<this.replicaUpSet.size(); i++)
                    {
                        pSocket.send(new DatagramPacket(checkpointData,checkpointData.length,
                                    InetAddress.getByName("localhost"),replicaUpSet.get(i)));
                    }
                    //writeLog(Constants.COMMIT+";"+incomingPayload.view+";"+incomingPayload.sequenceNum+";"+
                    //incomingPayload.digest+";");
            }
            catch (UnknownHostException e)
            {
                    e.printStackTrace();
            }
            catch (IOException e)
            {
                    e.printStackTrace();
            }
        }
       
    public void processCheckpoint(Payload incomingPayload)
    {
    	boolean verified = validateCheckpoint(incomingPayload);
    	if(! verified)
    	{
    		return;
    	}
    	if(!isStable(incomingPayload))
    	{
    		return;
    	}
    	
    	 this.lowWaterMark = incomingPayload.sequenceNum;
    	 this.highWaterMark = lowWaterMark + 10;
    	 lastStableCheckpoint = incomingPayload.sequenceNum;
    	 //delete from log
    	 Iterator<String> it = listLog.iterator();
 		while (it.hasNext())
 		{
 			String logIt = (String) it.next();
 			if (Integer.parseInt(logIt.substring(0,1))< incomingPayload.sequenceNum)
 			{
 				it.remove();
 			}
 			
 		}
 		writeLog(listLog);

    }
	
	public void processPrepare(Payload incomingPayload)
	{
		
		boolean verified = validatePrepare(incomingPayload);
		if(!verified)
		{
			return;
		}

		if(!isPrepared(incomingPayload))
		{
			return;
		}
		//now send commits to all 
		
		byte[] commitData = (new PayloadBuilder()).buildCommit(this.view,incomingPayload.sequenceNum,
		this.pid,this.myKeyPair,incomingPayload.digest,incomingPayload.clientPayload);

		Scanner scanInp = new Scanner(System.in);
		String byzMesgType = "";
		String byzBehave = "";
    	if(isByzantine)
    	{
    		System.out.print("PID: "+pid+ " - Behave Byzantine to disrupt view?: ");
    		
    		byzBehave = scanInp.next();
    		System.out.println("");
    		if(byzBehave.equals(Constants.YES))
    		{
    			System.out.print("PID: "+pid+ " - Byzantine message to send? ");
    			
    			byzMesgType = scanInp.next();
    			System.out.println("");
    			if(byzMesgType.equals(Constants.NEWVIEW))
    			{
    				commitData = (new PayloadBuilder()).buildNewView(this.view, this.pid, this.myKeyPair, null, 
    						null, lastStableCheckpoint, 0);
    			}
    			else if(byzMesgType.equals(Constants.VIEWCHANGE))
    			{
    				commitData = (new PayloadBuilder()).buildViewChange(view, pid, this.myKeyPair, lastStableCheckpoint, 
    						null, null, 0);
    			}
    		}
    	}
    	else
    	{
    		System.out.println("PID: "+pid+" - Sending COMMIT with sequence number: "+incomingPayload.sequenceNum+" for client request: "+
					incomingPayload.request+" to PEERS");
    		scanInp.nextLine();
    	}

		
		try
		{
			for( int i=0; i<this.replicaUpSet.size(); i++)
			{
				if(!(byzBehave.equals(Constants.YES) && i==this.pid))
				{
					pSocket.send(new DatagramPacket(commitData,commitData.length,
			    		InetAddress.getByName("localhost"),replicaUpSet.get(i)));
				}
			}
			
			listLog.add(incomingPayload.sequenceNum+","+Constants.COMMIT+","+incomingPayload.view+","+incomingPayload.digest+","+pid);
	   		writeLog(listLog);
		}
		catch (UnknownHostException e) 
		{
			e.printStackTrace();
		}
		catch (IOException e) 
		{
			e.printStackTrace();
		}
	}
	
	public ArrayList<Payload> calcremainingPrePrepare(int lastStableCheckpoint,int max)
	{
		return (new ArrayList<Payload>(listClientPayload.subList(lastStableCheckpoint, max)));
	}
	
	public boolean processViewChange(Payload incomingPayload) 
	{
		boolean verified = validateViewChange(incomingPayload);
		if(!verified)
		{
			return false;
		}

		if(!isViewChanged(incomingPayload))
		{
			return false;
		}
		//now send view change to all
		viewChangePayload = incomingPayload;
		//System.out.println("DEBUG sending view change to all");
   
		return true;
	}
		
	

	
	public void processNewView(Payload incomingPayload) 
	{
		// TODO Auto-generated method stub
		boolean verified = validateNewView(incomingPayload);
		if(!verified)
		{
			return;
		}
		
		// send prepares
		
		
		this.waitingInViewChange = false;

		/*try
		{
			for(int j=incomingPayload.lastStableCheckpoint;j<=incomingPayload.maxSeqNum;j++)
			{
				byte[] prepareData = (new PayloadBuilder()).buildPrepare(this.view,j,
						this.pid,this.myKeyPair,incomingPayload.digest,listClientPayload.get(j));
				System.out.println("DEBUG sending Prepares to all");
				for( int i=0; i<this.replicaUpSet.size(); i++)
				{
				    pSocket.send(new DatagramPacket(prepareData,prepareData.length,
				    		InetAddress.getByName("localhost"),replicaUpSet.get(i)));
				}
				listPrepareSent.add(view+incomingPayload.sequenceNum+incomingPayload.digest);
				writeLog(Constants.PREPARE+";"+incomingPayload.view+";"+incomingPayload.sequenceNum+";"+
				incomingPayload.digest+";");
			}
		}
		catch (UnknownHostException e) 
		{
			e.printStackTrace();
		}
		catch (IOException e) 
		{
			e.printStackTrace();
		}*/

		
	}

	
	public ArrayList<String> calcProofOfCheck(int lastStableCheckpoint)
	{
		ArrayList<String> listProofCheckpoint = new ArrayList<String>();
		for (int i=0;i<listCheckpointAccepted.size();i++)
		{
			if(listCheckpointAccepted.get(i).subSequence(1,1).equals(lastStableCheckpoint))
			{
				listProofCheckpoint.add(listCheckpointAccepted.get(i));
			}
		}
		return listProofCheckpoint;
	}
	
	public ArrayList<ArrayList<String>> calcProofOfPrepare(int lastStableCheckpoint,ArrayList<Integer> listSeqGreater)
	{
		ArrayList<ArrayList<String>> listProofPrepare = new ArrayList<ArrayList<String>>();
		ArrayList<String> listProofEachMesg = new ArrayList<String>();
		ArrayList<Integer> listCountSeq = new ArrayList<Integer>();
		for(int i=0;i<Constants.TOT_SEQNUM;i++)
		{
			listCountSeq.add(Constants.ZERO);
		}
		int newSeqNum =0 ;
		for (int i=0;i<listPrepareAccepted.size();i++)
		{
			newSeqNum = Integer.parseInt(listPrepareAccepted.get(i).substring(1,2));
			if( newSeqNum > lastStableCheckpoint)
			{
				int currCnt = listCountSeq.get(newSeqNum);
				listCountSeq.set(newSeqNum,currCnt+1);
				if(listCountSeq.get(newSeqNum) >= 2*Constants.FAULT_TOL)
				{
					if(listSeqGreater.indexOf(newSeqNum) != -1)
					{
						listSeqGreater.add(newSeqNum);
					}
				}
				
			}
		}
		
		for(int i=0; i<listSeqGreater.size();i++)
		{
			for(int j=0;j<listPrepareAccepted.size();j++)
			{
				newSeqNum = Integer.parseInt(listPrepareAccepted.get(j).substring(1,2));	
				if(newSeqNum == listSeqGreater.get(i))
				{
					listProofEachMesg.add(listPrepareAccepted.get(i));
				}
			}
			listProofPrepare.add(listProofEachMesg);
			listProofEachMesg.clear();
		}
		
		return listProofPrepare;
	}
	
	
   public int playPrimary(boolean viewChange)
    {
	   System.out.println("PID: "+pid+ " - PLAYING PRIMARY");
        // receive
        byte[] buf = new byte[Constants.PAYLOAD_SIZE];
        DatagramPacket incomingPacket = new DatagramPacket(buf,buf.length);
        int stateMachineState = Constants.CONTINUE_PROCESS;
        boolean verified;
        while (stateMachineState == Constants.CONTINUE_PROCESS)
        {
            try 
            {
            	if(!viewChange)
            	{
	                Arrays.fill(incomingPacket.getData(), 0,Constants.PAYLOAD_SIZE,(byte)0);
	                pSocket.setSoTimeout(peerTimeout);
	                System.out.println("PID: "+pid+ " - Waiting in primary");
	                pSocket.receive(incomingPacket);
            	}
                else
                {
                	viewChange = false;
            		ArrayList<Payload> remainingPrePrepares = new ArrayList<Payload>();
            		remainingPrePrepares = calcremainingPrePrepare(viewChangePayload.lastStableCheckpoint,viewChangePayload.maxSeqNum);
            		byte[] newViewData = (new PayloadBuilder()).buildNewView(viewChangePayload.view,this.pid,this.myKeyPair,
            				listViewChangeAccepted,remainingPrePrepares,lastStableCheckpoint,viewChangePayload.maxSeqNum);

            		try
            		{
            			
            			for( int i=0; i<this.replicaUpSet.size() ; i++)
            			{
            				if(i != this.pid)
            				{
            				
            					System.out.println("PID: "+pid+ " - Sending NEWVIEW message to: "+i);
            					pSocket.send(new DatagramPacket(newViewData,newViewData.length,
            			    		InetAddress.getByName("localhost"),replicaUpSet.get(i)));
            				}
            			}
            			this.peerTimeout = 0;
            			//writeLog(Constants.COMMIT+";"+incomingPayload.view+";"+incomingPayload.sequenceNum+";"+
            			//incomingPayload.digest+";");
            		}
            		catch (UnknownHostException e) 
            		{
            			e.printStackTrace();
            		}
            		catch (IOException e) 
            		{
            			e.printStackTrace();
            		}
            		continue;
                }
            }
            catch (IOException e) 
            {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            }
            Payload incomingPayload = (new PayloadBuilder()).deserializePayload(incomingPacket);
            if (incomingPayload.packetType.equals(Constants.REQUEST))
            {
            	System.out.println("PID: "+pid+ " - Received REQUEST from "+incomingPayload.senderPid);
                PublicKey clientPublicKey = addPublicKey(Constants.CLIENT,incomingPayload.senderPid);
                verified = Utilities.verifySignedData(incomingPayload, clientPublicKey);
				if(verified)
				{
					byte[] prePrepareData = (new PayloadBuilder()).buildPrePrepare(this.view,this.currSequenceNum++,
					this.pid,this.myKeyPair,Utilities.hashMyData(incomingPayload.payloadByteArray()), incomingPayload);
	                try
	                {
	                	clientList.get(incomingPayload.senderPid);
	                }
	                catch (IndexOutOfBoundsException ioobe) 
	                {
	                	
	                	clientList.add(incomingPayload.clientPort);
	                	lastClientResponse.add(Constants.EMPTY_STR);
	                	listClientPayload.add(incomingPayload);
					}
	                if (lastClientResponse.get(incomingPayload.senderPid).endsWith(incomingPayload.request))
	                {
	                	Payload clientResponsePayload = new  Payload();
	                	clientResponsePayload.clientPayload = incomingPayload;
	                	sendResponseToClient(clientResponsePayload);
	                	continue;
	                }
	                
	                
	        		Scanner scanInp = new Scanner(System.in);
	        		String byzMesgType = "";
	        		int unluckyNode = -1;
	            	if(isByzantine)
	            	{
	            		System.out.print("PID: "+pid+ " - Behave Byzantine?: ");
	            		String byzBehave = scanInp.next();
	            		System.out.println("");
	            		if(byzBehave.equals(Constants.YES))
	            		{
	            			System.out.print("PID: "+pid+ " - Enter Byzantine message to be sent: ");
	            			byzMesgType = scanInp.next();
	            			System.out.println("");
	            			
	            			if(byzMesgType.equals(Constants.WATERMARK))
	            			{
	            				System.out.println("PID: "+pid+ " - Sending a incorrect sequence number");
	            				System.out.print("PID: "+pid+ " - Enter incorrect sequence number to be sent: ");
		            			byzMesgType = scanInp.next();
		            			System.out.println("");
	            				prePrepareData = (new PayloadBuilder()).buildPrePrepare(this.view,Integer.parseInt(byzMesgType),
	            						this.pid,this.myKeyPair,Utilities.hashMyData(incomingPayload.payloadByteArray()), incomingPayload);
	            			}
	            			else if(byzMesgType.equals(Constants.HIGHVIEW))
	            			{
	            				System.out.println("PID: "+pid+ " - Sending invalid view number");
	            				System.out.print("PID: "+pid+ " - Enter incorrect view number to be sent: ");
		            			byzMesgType = scanInp.next();
		            			System.out.println("");
	            				prePrepareData = (new PayloadBuilder()).buildPrePrepare(Integer.parseInt(byzMesgType),this.currSequenceNum,
	            						this.pid,this.myKeyPair,Utilities.hashMyData(incomingPayload.payloadByteArray()), incomingPayload);
	            			}
	            			else if(byzMesgType.equals(Constants.B_REQUEST))
	            			{
	            				System.out.println("PID: "+pid+ " - Changing incoming client request");
	            				System.out.print("PID: "+pid+ " - Enter Byzantine message to be sent: ");
	            				incomingPayload.request = scanInp.next();
		            			System.out.println("");
	            				prePrepareData = (new PayloadBuilder()).buildPrePrepare(this.view,this.currSequenceNum,
	            						this.pid,this.myKeyPair,Utilities.hashMyData(incomingPayload.payloadByteArray()), incomingPayload);
	            			}
	            			else if(byzMesgType.equals(Constants.SOME))
	            			{
	            				System.out.println("PID: "+pid+ " - Not sending PRE-PREPARE to node 2");
	            				unluckyNode = 2;
	            			}

	            		}
	            	}
	            	else
	            	{
	            		System.out.println("PID: "+pid+ " - Sending PRE-PREPARE with sequence number: "+(this.currSequenceNum-1)+" for client request: "+
								incomingPayload.request+" to PEERS ");
	            		scanInp.nextLine();
	            	}

	                
	                try
					{
						
						for( int i=0; i<replicaUpSet.size(); i++)
						{
							if(i!=unluckyNode)
							{
								
								pSocket.send(new DatagramPacket(prePrepareData,prePrepareData.length,InetAddress.getByName("localhost"),
									replicaUpSet.get(i)));
								
							}
		                }
						logItem = (this.currSequenceNum-1)+","+Constants.PREPREPARE+","+view+","+Utilities.hashMyData(incomingPayload.payloadByteArray());
						listLog.add(logItem);
						writeLog(listLog);
						
						String prePrepareCheck = view+""+(this.currSequenceNum-1)+Utilities.hashMyData(incomingPayload.payloadByteArray());
						listPrePrepareAccepted.add(prePrepareCheck);
					}
					catch (UnknownHostException e) 
					{
                		// TODO Auto-generated catch block
            			e.printStackTrace();
					} catch (IOException e) 
					{
                		// TODO Auto-generated catch block
                        e.printStackTrace();
        			}
				}
            }
            else if (incomingPayload.packetType.equals(Constants.PREPARE))
            {
            	System.out.println("PID: "+pid+ " - Received a PREPARE from replica "+incomingPayload.senderPid);
            	processPrepare(incomingPayload);
            }
			else if(incomingPayload.packetType.equals(Constants.COMMIT))
			{
				System.out.println("PID: "+pid+ " - Received a COMMIT from replica "+incomingPayload.senderPid);
				processCommit(incomingPayload);
			}
			else if (incomingPayload.packetType.equals(Constants.VIEWCHANGE))
			{
				System.out.println("PID: "+pid+ " - Received a VIEWCHANGE from replica "+incomingPayload.senderPid+" --WHILE I AM ALIVE--. REJECTING!");
				continue;
			}
			else if (incomingPayload.packetType.equals(Constants.NEWVIEW))
			{
				System.out.println("PID: "+pid+ " - Received a NEWVIEW from "+incomingPayload.senderPid+"--WHILE I AM ALIVE--. REJECTING!");
				continue;
			}
			else if(incomingPayload.packetType.equals(Constants.CHECKPOINT))
			{
				System.out.println("PID: "+pid+ " - Received a CHECKPOINT from "+incomingPayload.senderPid);
				processCheckpoint(incomingPayload);
			}
            
        }
        return stateMachineState;
    }
	
    public int playBackup()
    {
    	System.out.println("PID: "+pid+ " - PLAYING BACKUP");
        byte[] buf = new byte[Constants.PAYLOAD_SIZE];
        DatagramPacket incomingPacket = new DatagramPacket(buf,buf.length);
        int stateMachineState = Constants.CONTINUE_PROCESS;
        boolean verified;
        while (stateMachineState == Constants.CONTINUE_PROCESS)
        {
            try 
            {

                Arrays.fill(incomingPacket.getData(), 0,Constants.PAYLOAD_SIZE,(byte)0);
                pSocket.setSoTimeout(peerTimeout);
                System.out.println("PID: "+pid+ " - Waiting in backup");
                pSocket.receive(incomingPacket);
            }
            catch (SocketTimeoutException ste) 
            {
				// TODO: handle exception
            	System.out.println("PID: "+this.pid+"  Timed OUT on primary :"+primaryPid);
            	waitingInViewChange = true;
            	peerTimeout = 0;
            	// changing view
            	//this.replicaUpSet.set(this.primaryPid, Constants.DEAD_PROCESS);
            	// multicast view change
            	ArrayList<String> listProofCheckpoint  = new ArrayList<String>();
            	ArrayList<ArrayList<String>> listProofPrepare = new ArrayList<ArrayList<String>>();
            	listProofCheckpoint = calcProofOfCheck(lastStableCheckpoint);
            	ArrayList<Integer> listSeqGreater = new ArrayList<Integer>();
            	listProofPrepare = calcProofOfPrepare(lastStableCheckpoint,listSeqGreater);
            	int max = 0;
            	for(int i=0;i<listSeqGreater.size();i++)
            	{
            		if(listSeqGreater.get(i) > max)
            		{
            			max = listSeqGreater.get(i);
            		}
            	}
            	
            	byte[] viewChangeData = (new PayloadBuilder()).buildViewChange(this.view,this.pid,this.myKeyPair,this.lastStableCheckpoint,
            			listProofCheckpoint,listProofPrepare,max);
            	for( int i=0; i<this.replicaUpSet.size(); i++)
				{
				    try {
				    	
				    	System.out.println("PID: "+pid+ " - Sending a VIEWCHANGE to replica "+i);
						pSocket.send(new DatagramPacket(viewChangeData,viewChangeData.length,
								InetAddress.getByName("localhost"),replicaUpSet.get(i)));
					} catch (UnknownHostException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
            	continue;
			}
            catch (IOException e) 
            {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            Payload incomingPayload = (new PayloadBuilder()).deserializePayload(incomingPacket);
            
            if (incomingPayload.packetType.equals(Constants.PREPREPARE))
            {
            	System.out.println("PID: "+pid+ " - Received a PRE-PREPARE from "+incomingPayload.senderPid+" for client: "+incomingPayload.clientPayload.senderPid);
            	if(waitingInViewChange)
            	{
            		continue;
            	}
            	verified = validatePrePrepare(incomingPayload);
                       
            	if(!verified)
				{
					continue;
				}
				else
				{
					//now send prepares to all 
					//PREPARE format: "PREPARE, view, seq, digest,myPid"
					try
	                {
	                	clientList.get(incomingPayload.clientPayload.senderPid);
	                }
	                catch (IndexOutOfBoundsException ioobe) 
	                {
	                	
	                	clientList.add(incomingPayload.clientPayload.clientPort);
	                	lastClientResponse.add(Constants.EMPTY_STR);
	                	listClientPayload.add(incomingPayload);
					}
	                
					byte[] prepareData = (new PayloadBuilder()).buildPrepare(this.view,incomingPayload.sequenceNum,
					this.pid,this.myKeyPair,incomingPayload.digest,incomingPayload.clientPayload);

					System.out.println("PID: "+pid+ " - Sending PREPARE with sequence number: "+incomingPayload.sequenceNum+" for client request: "+
							incomingPayload.request+"to PEERS");
					new Scanner(System.in).nextLine();
					try
					{
					
						for( int i=0; i<this.replicaUpSet.size(); i++)
						{
							if(this.pid != primaryPid)
							{
								
								pSocket.send(new DatagramPacket(prepareData,prepareData.length,
						    		InetAddress.getByName("localhost"),replicaUpSet.get(i)));
							}
						}
						listLog.add(incomingPayload.sequenceNum+","+Constants.PREPARE+","+incomingPayload.view+","+incomingPayload.digest+","+pid);
						writeLog(listLog);
					}
					catch (UnknownHostException e) 
					{
						e.printStackTrace();
					}
					catch (IOException e) 
					{
						e.printStackTrace();
					}
				}
		
            }
            else if (incomingPayload.packetType.equals(Constants.PREPARE))
            {
            	System.out.println("PID: "+pid+ " - Received a PREPARE from "+incomingPayload.senderPid);
            	if(waitingInViewChange)
            	{
            		continue;
            	}
            	processPrepare(incomingPayload);

            }
			else if(incomingPayload.packetType.equals(Constants.COMMIT))
			{
				System.out.println("PID: "+pid+ " - Received a COMMIT from "+incomingPayload.senderPid);
            	if(waitingInViewChange)
            	{
            		continue;
            	}
				processCommit(incomingPayload);
			}

			else if(incomingPayload.packetType.equals(Constants.VIEWCHANGE))
			{
				System.out.println("PID: "+pid+ " - Received a VIEWCHANGE from "+incomingPayload.senderPid);
				if(processViewChange(incomingPayload))
				{
					
					peerTimeout = 0;
					this.view = this.view + 1;
					System.out.println("PID: "+pid+ " - I have changed my view to "+this.view);
					return Constants.VIEW_CHANGE;
				}
			}
			else if(incomingPayload.packetType.equals(Constants.NEWVIEW))
			{
				System.out.println("PID: "+pid+ " - Received a NEWVIEW from "+incomingPayload.senderPid);
				processNewView(incomingPayload);
			}
			else if(incomingPayload.packetType.equals(Constants.CHECKPOINT))
			{
				System.out.println("PID: "+pid+ " - Received a CHECKPOINT from "+incomingPayload.senderPid);
				processCheckpoint(incomingPayload);
			}

			else if(incomingPayload.packetType.equals(Constants.REQUEST))
			{
				System.out.println("PID: "+pid+ " - Received a REQUEST from "+incomingPayload.senderPid);
            	if(waitingInViewChange)
            	{
            		continue;
            	}
				if (lastClientResponse.size()!=0 && lastClientResponse.get(incomingPayload.senderPid).endsWith(incomingPayload.request))
	            {
					Payload clientResponsePayload = new  Payload();
                	clientResponsePayload.clientPayload = incomingPayload;
                	sendResponseToClient(clientResponsePayload);
	            }
			}
        }
        return stateMachineState;
    }
  
	public int playClient()
    {
        System.out.println("PID: "+pid+" - PLAYING CLIENT");
        int stateMachineState = Constants.CONTINUE_PROCESS;
        while(stateMachineState != Constants.EXIT_MACH)
        {
	        long reqTimestamp = System.currentTimeMillis();
	        Scanner userIp = new Scanner(System.in);
	        System.out.print("Enter the message to be sent: ");
	        String request = clientAllReqIndx+userIp.next();
	        byte[] packetData = (new PayloadBuilder()).buildRequest(request,reqTimestamp,this.pid,this.myKeyPair,this.port);
	        // send here
	        try 
	        {
	            System.out.println("PID: "+pid+" - Sending request "+request+ " to primary: "+primaryPid);
	            pSocket.send(new DatagramPacket(packetData,packetData.length,InetAddress.getByName("localhost"),replicaUpSet.get(primaryPid)));
	        } 
	        catch (UnknownHostException e) 
	        {
	            // TODO Auto-generated catch block
	            e.printStackTrace();
	        } 
	        catch (IOException e) 
	        {
	            // TODO Auto-generated catch block
	            e.printStackTrace();
	        }
	        // receive
	        byte[] buf = new byte[Constants.PAYLOAD_SIZE];
	        DatagramPacket incomingPacket = new DatagramPacket(buf,buf.length);
	        stateMachineState = Constants.CONTINUE_PROCESS;
	        while(stateMachineState == Constants.CONTINUE_PROCESS)
	        {
	            try 
	            {
	                Arrays.fill(incomingPacket.getData(), 0,Constants.PAYLOAD_SIZE,(byte)0);
	                pSocket.setSoTimeout(peerTimeout);
	                pSocket.receive(incomingPacket);
	            }
	            catch (SocketTimeoutException ste) 
	            {
	            	System.out.println("PID: "+pid+" - Timed out on current primary: "+primaryPid);
	                
	                view = view + 1;
	            	for( int i=0; i<this.replicaUpSet.size(); i++)
					{
	                    try 
	                    {
	                    	if(i != primaryPid)
	                    	{
	                    		pSocket.send(new DatagramPacket(packetData,packetData.length,
										InetAddress.getByName("localhost"),replicaUpSet.get(i)));
								System.out.println("PID: "+pid+" - Sending request "+request+ " to backup "+i);
	                    	}
							
						} catch (UnknownHostException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
	            	this.primaryPid = view%(totalPeers);
	            	
	            	peerTimeout = 0;
	            	continue;
				}
	            catch (IOException e) 
	            {
	                    // TODO Auto-generated catch block
	                    e.printStackTrace();
	            }
	            Payload incomingPayload = (new PayloadBuilder()).deserializePayload(incomingPacket);
	            int replyFound = 0;
	            String replyCheck = reqTimestamp+""+request+incomingPayload.reply; 
	            
	            if (incomingPayload.packetType.equals(Constants.REPLY))
	            {	
	            	System.out.println("PID: "+pid+ " - Received reply "+incomingPayload.reply + " from process "+incomingPayload.senderPid);
	            	if (validateReply(incomingPayload, request,reqTimestamp))
	            	{
	            		
	            		for (int i=0;i<listReplyAccepted.size();i++)
	            		{
							if(listReplyAccepted.get(i).substring(1).equals(replyCheck))
	            			{
								replyFound++;
	            			}
	            		}
	            	}
	            }
	           
	            if(replyFound == Constants.FAULT_TOL+1)
	            {
	            	System.out.println("PID: "+pid+" - Got f+1 matching replies. Final Result is "+ incomingPayload.reply);
	            	this.primaryPid = incomingPayload.view%(totalPeers);
	            	
	            	stateMachineState = Constants.REPLY_GOT;
	            	clientAllReqIndx++;
	            }
	        }
        }
        return stateMachineState;
    }
    
    public void writeLog(ArrayList<String> msg)
    {
        try
        {
            BufferedWriter bw = new BufferedWriter(new FileWriter(logFile,false));
            for(int i=0;i<msg.size();i++)
            {
            	bw.write(msg.get(i)+"\n");	
            }
            
            
            bw.flush();
        }
        catch(IOException io)
        {
            System.err.println(io.getMessage());
        }
    }

}

